/* PeerState - Keeps track of the Peer state through connection callbacks.
   Copyright (C) 2003 Mark J. Wielaard

   This file is part of Snark.
   
   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2, or (at your option)
   any later version.
 
   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.
 
   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software Foundation,
   Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
*/

package org.klomp.snark;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.HashSet;

class PeerState
{
  final Peer peer;
  final PeerListener listener;
  final MetaInfo metainfo;

  // Interesting and choking describes whether we are interested in or
  // are choking the other side.
  boolean interesting = false;
  boolean choking = true;

  // Interested and choked describes whether the other side is
  // interested in us or choked us.
  boolean interested = false;
  boolean choked = true;

  // Package local for use by Peer.
  long downloaded;
  long uploaded;

  BitField bitfield;

  // Package local for use by Peer.
  final PeerConnectionIn in;
  final PeerConnectionOut out;

  // Outstanding request
  private final List outstandingRequests = new ArrayList();
  private Request lastRequest = null;

  // If we have te resend outstanding requests (true after we got choked).
  private boolean resend = false;

  private final static int MAX_PIPELINE = 5;
  private final static int PARTSIZE = 16384; // 16K

  PeerState(Peer peer, PeerListener listener, MetaInfo metainfo,
	    PeerConnectionIn in, PeerConnectionOut out)
  {
    this.peer = peer;
    this.listener = listener;
    this.metainfo = metainfo;

    this.in = in;
    this.out = out;
  }

  // NOTE Methods that inspect or change the state synchronize (on this).

  void keepAliveMessage()
  {
    if (Snark.debug >= Snark.DEBUG)
      Snark.debug(peer + " rcv alive", Snark.DEBUG);
    /* XXX - ignored */
  }

  void chokeMessage(boolean choke)
  {
    if (Snark.debug >= Snark.DEBUG)
      Snark.debug(peer + " rcv " + (choke ? "" : "un") + "choked",
		  Snark.DEBUG);

    choked = choke;
    if (choked)
      resend = true;

    listener.gotChoke(peer, choke);

    if (!choked && interesting)
      request();
  }

  void interestedMessage(boolean interest)
  {
    if (Snark.debug >= Snark.DEBUG)
      Snark.debug(peer + " rcv " + (interest ? "" : "un")
		  + "interested", Snark.DEBUG);
    interested = interest;
    listener.gotInterest(peer, interest);
  }

  void haveMessage(int piece)
  {
    if (Snark.debug >= Snark.DEBUG)
      Snark.debug(peer + " rcv have(" + piece + ")", Snark.DEBUG);
    // Sanity check
    if (piece < 0 || piece >= metainfo.getPieces())
      {
	// XXX disconnect?
	if (Snark.debug >= Snark.INFO)
	  Snark.debug("Got strange 'have: " + piece + "' message from " + peer,
		      + Snark.INFO);
	return;
      }

    synchronized(this)
      {
	// Can happen if the other side never send a bitfield message.
	if (bitfield == null)
	  bitfield = new BitField(metainfo.getPieces());

	bitfield.set(piece);
      }

    if (listener.gotHave(peer, piece))
      setInteresting(true);
  }

  void bitfieldMessage(byte[] bitmap)
  {
    synchronized(this)
      {
	if (Snark.debug >= Snark.DEBUG)
	  Snark.debug(peer + " rcv bitfield", Snark.DEBUG);
	if (bitfield != null)
	  {
	    // XXX - Be liberal in what you except?
	    if (Snark.debug >= Snark.INFO)
	      Snark.debug("Got unexpected bitfield message from " + peer,
			  Snark.INFO);
	    return;
	  }
	
	// XXX - Check for weird bitfield and disconnect?
	bitfield = new BitField(bitmap, metainfo.getPieces());
      }
    setInteresting(listener.gotBitField(peer, bitfield));
  }

  void requestMessage(int piece, int begin, int length)
  {
    if (Snark.debug >= Snark.DEBUG)
      Snark.debug(peer + " rcv request("
		  + piece + ", " + begin + ", " + length + ") ",
		  Snark.DEBUG);
    if (choking)
      {
	if (Snark.debug >= Snark.INFO)
	  Snark.debug("Request received, but choking " + peer, Snark.INFO);
	return;
      }

    // Sanity check
    if (piece < 0
	|| piece >= metainfo.getPieces()
	|| begin < 0
	|| begin > metainfo.getPieceLength(piece)
	|| length <= 0
	|| length > 4*PARTSIZE)
      {
	// XXX - Protocol error -> disconnect?
	if (Snark.debug >= Snark.INFO)
	  Snark.debug("Got strange 'request: " + piece
		      + ", " + begin
		      + ", " + length
		      + "' message from " + peer,
		      Snark.INFO);
	return;
      }

    byte[] pieceBytes = listener.gotRequest(peer, piece);
    if (pieceBytes == null)
      {
	// XXX - Protocol error-> diconnect?
	if (Snark.debug >= Snark.INFO)
	  Snark.debug("Got request for unknown piece: " + piece, Snark.INFO);
	return;
      }

    // More sanity checks
    if (begin >= pieceBytes.length || begin + length > pieceBytes.length)
      {
	// XXX - Protocol error-> disconnect?
	if (Snark.debug >= Snark.INFO)
	  Snark.debug("Got out of range 'request: " + piece
		      + ", " + begin
		      + ", " + length
		      + "' message from " + peer,
		      Snark.INFO);
	return;
      }

    if (Snark.debug >= Snark.DEBUG)
      Snark.debug("Sending (" + piece + ", " + begin + ", "
		  + length + ")" + " to " + peer, Snark.DEBUG);
    out.sendPiece(piece, begin, length, pieceBytes);

    // Tell about last subpiece delivery.
    if (begin + length == pieceBytes.length)
      if (Snark.debug >= Snark.DEBUG)
	Snark.debug("Send p" + piece + " " + peer,
		    Snark.DEBUG);
  }

  /**
   * Called when some bytes have left the outgoing connection.
   * XXX - Should indicate whether it was a real piece or overhead.
   */
  void uploaded(int size)
  {
    uploaded += size;
    listener.uploaded(peer, size);
  }

  /**
   * Called when a partial piece request has been handled by
   * PeerConnectionIn.
   */
  void pieceMessage(Request req)
  {
    int size = req.len;
    downloaded += size;
    listener.downloaded(peer, size);

    // Last chunk needed for this piece?
    if (getFirstOutstandingRequest(req.piece) == -1)
      {
	if (listener.gotPiece(peer, req.piece, req.bs))
	  {
	    if (Snark.debug >= Snark.DEBUG)
	      Snark.debug("Got " + req.piece + ": " + peer, Snark.DEBUG);
	  }
	else
	  {
	    if (Snark.debug >= Snark.DEBUG)
	      Snark.debug("Got BAD " + req.piece + " from " + peer,
			  Snark.DEBUG);
	    // XXX ARGH What now !?!
	    downloaded = 0;
	  }
      }
  }

  synchronized private int getFirstOutstandingRequest(int piece)
  {
    for (int i = 0; i < outstandingRequests.size(); i++)
      if (((Request)outstandingRequests.get(i)).piece == piece)
	return i;
    return -1;
  }

  /**
   * Called when a piece message is being processed by the incoming
   * connection. Returns null when there was no such request. It also
   * requeues/sends requests when it thinks that they must have been
   * lost.
   */
  Request getOutstandingRequest(int piece, int begin, int length)
  {
    if (Snark.debug >= Snark.DEBUG)
      Snark.debug("getChunk("
		  + piece + "," + begin + "," + length + ") "
		  + peer, Snark.DEBUG);

    int r = getFirstOutstandingRequest(piece);

    // Unrequested piece number?
    if (r == -1)
      {
	if (Snark.debug >= Snark.INFO)
	  Snark.debug("Unrequested 'piece: " + piece + ", "
		      + begin + ", " + length + "' received from "
		      + peer,
		      Snark.INFO);
	downloaded = 0; // XXX - punishment?
	return null;
      }

    // Lookup the correct piece chunk request from the list.
    Request req;
    synchronized(this)
      {
	req = (Request)outstandingRequests.get(r);
	while (req.piece == piece && req.off != begin
	       && r < outstandingRequests.size() - 1)
	  {
	    r++;
	    req = (Request)outstandingRequests.get(r);
	  }
	
	// Something wrong?
	if (req.piece != piece || req.off != begin || req.len != length)
	  {
	    if (Snark.debug >= Snark.INFO)
	      Snark.debug("Unrequested or unneeded 'piece: "
			  + piece + ", "
			  + begin + ", "
			  + length + "' received from "
			  + peer,
			  Snark.INFO);
	    downloaded = 0; // XXX - punishment?
	    return null;
	  }
	
	// Report missing requests.
	if (r != 0)
	  {
	    if (Snark.debug >= Snark.INFO)
	      System.err.print("Some requests dropped, got " + req
			       + ", wanted:");
	    for (int i = 0; i < r; i++)
	      {
		Request dropReq = (Request)outstandingRequests.remove(0);
		outstandingRequests.add(dropReq);
		// We used to rerequest the missing chunks but that mostly
		// just confuses the other side. So now we just keep
		// waiting for them. They will be rerequested when we get
		// choked/unchoked again.
		/*
		  if (!choked)
		  out.sendRequest(dropReq);
		*/
		if (Snark.debug >= Snark.INFO)
		  System.err.print(" " + dropReq);
	      }
	    if (Snark.debug >= Snark.INFO)
	      System.err.println(" " + peer);
	  }
	outstandingRequests.remove(0);
      }

    // Request more if necessary to keep the pipeline filled.
    addRequest();

    return req;

  }

  void cancelMessage(int piece, int begin, int length)
  {
    if (Snark.debug >= Snark.DEBUG)
      Snark.debug("Got cancel message ("
		  + piece + ", " + begin + ", " + length + ")",
		  Snark.DEBUG);
    out.cancelRequest(piece, begin, length);
  }

  void unknownMessage(int type, byte[] bs)
  {
    if (Snark.debug >= Snark.WARNING)
      Snark.debug("Warning: Ignoring unknown message type: " + type
		  + " length: " + bs.length, Snark.WARNING);
  }

  void havePiece(int piece)
  {
    if (Snark.debug >= Snark.DEBUG)
      Snark.debug("Tell " + peer + " havePiece(" + piece + ")", Snark.DEBUG);

    synchronized(this)
      {
	// Tell the other side that we are no longer interested in any of
	// the outstanding requests for this piece.
	if (lastRequest != null && lastRequest.piece == piece)
	  lastRequest = null;
	
	Iterator it = outstandingRequests.iterator();
	while (it.hasNext())
	  {
	    Request req = (Request)it.next();
	    if (req.piece == piece)
	      {
		it.remove();
		// Send cancel even when we are choked to make sure that it is
		// really never ever send.
		out.sendCancel(req);
	      }
	  }
      }
    
    // Tell the other side that we really have this piece.
    out.sendHave(piece);
    
    // Request something else if necessary.
    addRequest();
    
    synchronized(this)
      {
	// Is the peer still interesting?
	if (lastRequest == null)
	  setInteresting(false);
      }
  }

  // Starts or resumes requesting pieces.
  private void request()
  {
    // Are there outstanding requests that have to be resend?
    if (resend)
      {
	out.sendRequests(outstandingRequests);
	resend = false;
      }

    // Add/Send some more requests if necessary.
    addRequest();
  }

  /**
   * Adds a new request to the outstanding requests list.
   */
  private void addRequest()
  {
    boolean more_pieces = true;
    while (more_pieces)
      {
	synchronized(this)
	  {
	    more_pieces = outstandingRequests.size() < MAX_PIPELINE;
	  }
	
	// We want something and we don't have outstanding requests?
	if (more_pieces && lastRequest == null)
	  more_pieces = requestNextPiece();
	else if (more_pieces) // We want something
	  {
	    int pieceLength;
	    boolean isLastChunk;
	    synchronized(this)
	      {
		pieceLength = metainfo.getPieceLength(lastRequest.piece);
		isLastChunk = lastRequest.off + lastRequest.len == pieceLength;
	      }

	    // Last part of a piece?
	    if (isLastChunk)
	      more_pieces = requestNextPiece();
	    else
	      {
		synchronized(this)
		  {
		    int nextPiece = lastRequest.piece;
		    int nextBegin = lastRequest.off + PARTSIZE;
		    byte[] bs = lastRequest.bs;
		    int maxLength = pieceLength - nextBegin;
		    int nextLength = maxLength > PARTSIZE ? PARTSIZE
		                                          : maxLength;
		    Request req
		      = new Request(nextPiece, bs, nextBegin, nextLength);
		    outstandingRequests.add(req);
		    if (!choked)
		      out.sendRequest(req);
		    lastRequest = req;
		  }
	      }
	  }
      }

    if (Snark.debug >= Snark.DEBUG)
      Snark.debug(peer + " requests " + outstandingRequests, Snark.DEBUG);
  }

  // Starts requesting first chunk of next piece. Returns true if
  // something has been added to the requests, false otherwise.
  private boolean requestNextPiece()
  {
    // Check that we already know what the other side has.
    if (bitfield != null)
      {
	int nextPiece = listener.wantPiece(peer, bitfield);
	if (Snark.debug >= Snark.DEBUG)
	  Snark.debug(peer + " want piece " + nextPiece, Snark.DEBUG);
	synchronized(this)
	  {
	    if (nextPiece != -1
		&& (lastRequest == null || lastRequest.piece != nextPiece))
	      {
		int piece_length = metainfo.getPieceLength(nextPiece);
		byte[] bs = new byte[piece_length];
		
		int length = Math.min(piece_length, PARTSIZE);
		Request req = new Request(nextPiece, bs, 0, length);
		outstandingRequests.add(req);
		if (!choked)
		  out.sendRequest(req);
		lastRequest = req;
		return true;
	      }
	  }
      }

    return false;
  }

  synchronized void setInteresting(boolean interest)
  {
    if (Snark.debug >= Snark.DEBUG)
      Snark.debug(peer + " setInteresting(" + interest + ")", Snark.DEBUG);

    if (interest != interesting)
      {
	interesting = interest;
	out.sendInterest(interest);

	if (interesting && !choked)
	  request();
      }
  }

  synchronized void setChoking(boolean choke)
  {
    if (Snark.debug >= Snark.DEBUG)
      Snark.debug(peer + " setChoking(" + choke + ")", Snark.DEBUG);

    if (choking != choke)
      {
	choking = choke;
	out.sendChoke(choke);
      }
  }
}
